package com.shadowlight.hero_story.mq;

import com.alibaba.fastjson.JSONObject;
import com.shadowlight.hero_story.config.AllConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;

/**
 * @Description 消息队列生产者
 * @Author cz
 * @Date 2022/4/15
 */
@Slf4j
public final class MQProducer {

    private static DefaultMQProducer producer;

    private MQProducer() {
    }

    public static void init() {
        if (!AllConfig.ROCKETMQ_ENABLE) {
            log.debug("跳过消费者初始化!");
            return;
        }
        try {
            DefaultMQProducer producer = new DefaultMQProducer("hero_story", getAclRPCHook());
            producer.setNamesrvAddr(AllConfig.ROCKETMQ_NAMESRV_ADDR);
            producer.start();
            producer.setRetryTimesWhenSendFailed(3);
            MQProducer.producer = producer;
            log.info("MQ生产者启动成功");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    public static void sendMsg(String topic, Object obj) {
        if (StringUtils.isEmpty(topic) || obj == null) {
            throw new RuntimeException("发送消息必填项不能为空");
        }
        if (producer == null) {
            throw new RuntimeException("生产者尚未初始化");
        }
        Message msg = new Message();
        msg.setTopic(topic);
        msg.setBody(JSONObject.toJSONBytes(obj));

        try {
            producer.send(msg);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    //ACL相关key
    static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("heroStory", "heroStory6677@"));
    }
}
